package com.fc.test.compare;

import com.fc.test.common.conf.ImpalaConfig;
import com.fc.test.common.domain.BaseObject;
import com.fc.test.common.schema.Table;
import com.fc.test.model.auto.TConnection;
import com.fc.test.model.auto.TableConfig;
import com.fc.test.model.auto.TableLoggerWithBLOBs;
import com.fc.test.service.TableLoggerService;
import com.fc.test.util.ClassUtil;
import com.fc.test.util.DataSourceManager;
import com.fc.test.util.DateUtils;
import com.fc.test.util.SpringContextUtils;
import com.fc.test.util.StringUtils;
import com.google.common.base.Preconditions;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * Created by Liaopan on 2019/7/4.
 */
public abstract class AbstractMapCompare implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(AbstractMapCompare.class);

    private TConnection source;
    private TConnection target;

    protected DataSource sourceDataSource;
    protected DataSource targetDataSource;

    protected TableConfig table;

    protected Table sourceTable;
    protected Table targetTable;

    protected Class sourceClass;
    protected Class targetClass;

    public AbstractMapCompare(TableConfig tableConfig) {
        this.table = tableConfig;
    }

    private void initialize(){
        source = table.getSourceConfig();
        target = table.getTargetConfig();
        Preconditions.checkNotNull(source,"确认源连接信息是否存在？");
        Preconditions.checkNotNull(target,"确认目标连接源是否存在？");
        try {
            // mysql 获取元数据需要连接字符串带数据库
            if(source.getUrl().startsWith("jdbc:mysql")) {
                source.setUrl(source.getUrl() + "/" + StringUtils.substringBefore(table.getSourceTable(),"."));
            }
            if(target.getUrl().startsWith("jdbc:mysql")) {
                target.setUrl(target.getUrl() + "/" + StringUtils.substringBefore(table.getTargetTable(),"."));
            }
            sourceDataSource = DataSourceManager.getInstance(source.getDriver(), source.getUrl(),
                    source.getUser(), source.getPassword());
            targetDataSource = DataSourceManager.getInstance(target.getDriver(), target.getUrl(),
                    target.getUser(), target.getPassword());

            if ("hive_impala".equalsIgnoreCase(table.getSourceCode())) {
                LOG.debug("源连接：初始化Hive 源转换成 Impala数据连接...");
                ImpalaConfig impalaConfig = SpringContextUtils.getBean("impalaConfig");
                sourceDataSource = DataSourceManager.getInstance(impalaConfig.getDriverClassName(),
                        impalaConfig.getUrl(), impalaConfig.getUsername(), impalaConfig.getPassword());
                source.setUrl(impalaConfig.getUrl());
                source.setDriver(impalaConfig.getDriverClassName());
                source.setUser(impalaConfig.getUsername());
                source.setPassword(impalaConfig.getPassword());
            }

            if ("hive_impala".equalsIgnoreCase(table.getTargetCode())) {
                LOG.debug("目标连接：初始化Hive 源转换成 Impala数据连接...");
                ImpalaConfig impalaConfig = SpringContextUtils.getBean("impalaConfig");
                targetDataSource = DataSourceManager.getInstance(impalaConfig.getDriverClassName(),
                        impalaConfig.getUrl(), impalaConfig.getUsername(), impalaConfig.getPassword());
                target.setUrl(impalaConfig.getUrl());
                target.setDriver(impalaConfig.getDriverClassName());
                target.setUser(impalaConfig.getUsername());
                target.setPassword(impalaConfig.getPassword());
            }

            sourceTable = new Table(table.getSourceTable(), table.getPartitionColumn(),
                    sourceDataSource.getConnection());
            targetTable = new Table(table.getTargetTable(), table.getPartitionColumn(),
                    targetDataSource.getConnection());

            StringBuilder sourceQuery = new StringBuilder(sourceTable.getQuery());
            StringBuilder sourceQueryAll = new StringBuilder(sourceTable.getQueryAll());
            StringBuilder targetQuery = new StringBuilder(targetTable.getQuery());
            StringBuilder targetQueryAll = new StringBuilder(targetTable.getQueryAll());

            String patitionColumnName = table.getPartitionColumn();
            if(StringUtils.isNotEmpty(patitionColumnName) && null != table.getStartDate()
                    && null != table.getEndDate()) {

                StringBuilder condition = new StringBuilder(" AND ").append(patitionColumnName).append(" BETWEEN ")
                        .append("{1}").append(" AND ").append("{2}");
                sourceQuery.append(StringUtils.messageFormat(condition.toString()
                        ,StringUtils.messageFormat(getSourceDateTemplate(),DateUtils.format(table.getStartDate(),"yyyy-MM-dd")),
                                StringUtils.messageFormat(getSourceDateTemplate(),DateUtils.format(table.getEndDate(),"yyyy-MM-dd"))));
                sourceQueryAll.append(StringUtils.messageFormat(condition.toString()
                        ,StringUtils.messageFormat(getSourceDateTemplate(),DateUtils.format(table.getStartDate(),"yyyy-MM-dd")),
                                StringUtils.messageFormat(getSourceDateTemplate(),DateUtils.format(table.getEndDate(),"yyyy-MM-dd"))));
                targetQuery.append(StringUtils.messageFormat(condition.toString()
                        ,StringUtils.messageFormat(getTargetDateTemplate(),DateUtils.format(table.getStartDate(),"yyyy-MM-dd")),
                                StringUtils.messageFormat(getTargetDateTemplate(),DateUtils.format(table.getEndDate(),"yyyy-MM-dd"))));
                targetQueryAll.append(StringUtils.messageFormat(condition.toString()
                        ,StringUtils.messageFormat(getTargetDateTemplate(),DateUtils.format(table.getStartDate(),"yyyy-MM-dd")),
                                StringUtils.messageFormat(getTargetDateTemplate(),DateUtils.format(table.getEndDate(),"yyyy-MM-dd"))));
                sourceTable.setQuery(sourceQuery.toString());
                targetTable.setQuery(targetQuery.toString());
                sourceTable.setQueryAll(sourceQueryAll.toString());
                targetTable.setQueryAll(targetQueryAll.toString());
            }

            // 处理查询主键和时间的sql: 如果有配置sql， 直接使用配置的替换掉时间； 如果没有，还是使用上面的query
            if(StringUtils.isNotEmpty(table.getSourceSql())) {
                sourceTable.setQuery(StringUtils.messageFormat(table.getSourceSql(),
                        DateUtils.format(table.getStartDate(),"yyyy-MM-dd"),
                        DateUtils.format(table.getEndDate(),"yyyy-MM-dd")));
            }
            if(StringUtils.isNotEmpty(table.getTargetSql())) {
                targetTable.setQuery(StringUtils.messageFormat(table.getTargetSql(),
                        DateUtils.format(table.getStartDate(),"yyyy-MM-dd"),
                        DateUtils.format(table.getEndDate(),"yyyy-MM-dd")));
            }

            sourceClass = ClassUtil.forName(sourceTable.getUnqualifiedName(),
                    sourceTable.getColumns());
            targetClass = ClassUtil.forName(targetTable.getUnqualifiedName(),
                    targetTable.getColumns());

        } catch (Exception e) {
            LOG.error("初始化连接和Table失败", e);
        }
    }

    protected abstract CompareResultMap taskRun(String sourceQuery, String targetQuery) throws Exception;

    @Override
    public void run() {
        CompareResultMap resultMap = null;
        try {
            initialize();
            resultMap = taskRun(sourceTable.getQuery(), targetTable.getQuery());
            if(JobConstants.OnlyCount.DETAIL.getCode().equals(table.getOnlyCount())) {
                if (resultMap.isSuccessed() &&
                        resultMap.getLeft().size() == resultMap.getRight().size()) {
                    LOG.info("比对数据量一致，查找所有内容进行比较...");
                    //释放list内存占用
                    resultMap = null;
                    System.gc();
                    resultMap = taskRun(sourceTable.getQueryAll(), targetTable.getQueryAll());
                }
            }
        } catch (Exception e) {
            LOG.error("数据执行失败", e);
        }finally {
            postTaskRun(resultMap);
        }
    }

    /**
     * 比对完成后保存流水日志数据，更新主表的时间
     * @param resultList
     */
    private void postTaskRun(CompareResultMap resultList) {
        if(resultList == null) {
            return ;
        }
        int lSize = Optional.ofNullable(resultList.getLeft()).map(Map::size).orElse(-1);
        int rSize = Optional.ofNullable(resultList.getRight()).map(Map::size).orElse(-1);
        int status = (StringUtils.isEmpty(resultList.getlMessage()) && StringUtils.isEmpty(resultList.getrMessage())) ? 0 : 1;
        String message = status == 0 ? null : resultList.getlMessage() + "\n|" + resultList.getrMessage();
        final Gson GSON = new GsonBuilder()
                .excludeFieldsWithoutExposeAnnotation()
                .setDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSS")
                .create();
        Pair<List<Object>, List<Object>> listPair = resultList.leftCompareRight();
        String sourceData = listPair.getLeft().size() == 0 ? null : listPair.getLeft().size() + "|" + GSON.toJson(listPair.getLeft());
        String targetData = listPair.getRight().size() == 0 ? null : listPair.getRight().size() + "|" +GSON.toJson(listPair.getRight());

        TableLoggerWithBLOBs tableLogger = new TableLoggerWithBLOBs(null, table.getId(),
                lSize+"" , ""+rSize, table.getStartDate(),
                table.getEndDate(), new Date(), status, message,sourceData, targetData);
        TableLoggerService service = SpringContextUtils.getBean("tableLoggerService");
        service.insertSelective(tableLogger);
    }

    public String getSourceDateTemplate() {
        return getDateFunTemplate(source.getUrl());
    }

    public String getTargetDateTemplate() {
        return getDateFunTemplate(target.getUrl());
    }

    public String getDateFunTemplate(String rawUrl) {
        if (rawUrl == null) {
            return null;
        }

        if (rawUrl.startsWith("jdbc:derby:") || rawUrl.startsWith("jdbc:log4jdbc:derby:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:mysql:") || rawUrl.startsWith("jdbc:cobar:")
                || rawUrl.startsWith("jdbc:log4jdbc:mysql:")) {
            return "'{1}'";
        } else if (rawUrl.startsWith("jdbc:mariadb:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:oracle:") || rawUrl.startsWith("jdbc:log4jdbc:oracle:")) {
            return "to_date('{1}','yyyy-mm-dd')";
        } else if (rawUrl.startsWith("jdbc:microsoft:") || rawUrl.startsWith("jdbc:log4jdbc:microsoft:")) {
            return "CONVERT(date, '{1}', 23)";
        } else if (rawUrl.startsWith("jdbc:sqlserver:") || rawUrl.startsWith("jdbc:log4jdbc:sqlserver:")) {
            return "CONVERT(date, '{1}', 23)";
        } else if (rawUrl.startsWith("jdbc:sybase:Tds:") || rawUrl.startsWith("jdbc:log4jdbc:sybase:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:postgresql:") || rawUrl.startsWith("jdbc:log4jdbc:postgresql:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:edb:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:hsqldb:") || rawUrl.startsWith("jdbc:log4jdbc:hsqldb:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:odps:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:db2:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:sqlite:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:ingres:")) {
            return "ingres";
        } else if (rawUrl.startsWith("jdbc:h2:") || rawUrl.startsWith("jdbc:log4jdbc:h2:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:mckoi:")) {
            return "mckoi";
        } else if (rawUrl.startsWith("jdbc:cloudscape:")) {
            return "cloudscape";
        } else if (rawUrl.startsWith("jdbc:informix-sqli:") || rawUrl.startsWith("jdbc:log4jdbc:informix-sqli:")) {
            return "informix";
        } else if (rawUrl.startsWith("jdbc:timesten:")) {
            return "timesten";
        } else if (rawUrl.startsWith("jdbc:as400:")) {
            return "as400";
        } else if (rawUrl.startsWith("jdbc:sapdb:")) {
            return "sapdb";
        } else if (rawUrl.startsWith("jdbc:JSQLConnect:")) {
            return "JSQLConnect";
        } else if (rawUrl.startsWith("jdbc:JTurbo:")) {
            return "JTurbo";
        } else if (rawUrl.startsWith("jdbc:firebirdsql:")) {
            return "firebirdsql";
        } else if (rawUrl.startsWith("jdbc:interbase:")) {
            return "interbase";
        } else if (rawUrl.startsWith("jdbc:pointbase:")) {
            return "pointbase";
        } else if (rawUrl.startsWith("jdbc:edbc:")) {
            return "edbc";
        } else if (rawUrl.startsWith("jdbc:mimer:multi1:")) {
            return "mimer";
        } else if (rawUrl.startsWith("jdbc:dm:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:kingbase:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:gbase:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:xugu:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:log4jdbc:")) {
            return "";
        } else if (rawUrl.startsWith("jdbc:hive:")) {
            return "'{1}'";
        } else if (rawUrl.startsWith("jdbc:hive2:")) {
            return "'{1}'";
        } else if (rawUrl.startsWith("jdbc:phoenix:")) {
            return "'{1}'";
        } else if (rawUrl.startsWith("jdbc:elastic:")) {
            return "'{1}'";
        } else if (rawUrl.startsWith("jdbc:clickhouse:")) {
            return "'{1}'";
        } else {
            return "'{1}'";
        }
    }
}
